#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据验证模块
用于验证广电用户数据预处理后的质量
"""
import pandas as pd

def validate_data_quality(df):
    """
    验证广电用户数据质量
    
    参数:
    df: 待验证的DataFrame
    
    返回:
    包含验证结果的字典
    """
    results = {
        'is_valid': True,
        'message': '数据验证通过',
        'duplicate_users': [],
        'invalid_records': [],
        'total_records': len(df),
        'valid_records': len(df)
    }
    
    # 显示可用的列名，帮助调试
    print(f"验证数据列名: {list(df.columns)}")
    
    # 尝试识别关键列
    columns_map = {}
    
    # 识别用户编号列
    for col in ['phone_no', 'distinct_usermsg.phone_no', '用户编号', '电话号码']:
        if col in df.columns:
            columns_map['phone'] = col
            break
    
    # 识别产品名称列
    for col in ['sm_name', 'distinct_usermsg.sm_name', '产品名称', '套餐名称']:
        if col in df.columns:
            columns_map['product'] = col
            break
    
    # 识别时间列
    for col in ['run_time', 'distinct_usermsg.run_time', '时间', '运行时间']:
        if col in df.columns:
            columns_map['time'] = col
            break
    
    print(f"识别到的关键列: {columns_map}")
    
    # 1. 检查用户编号唯一性
    if 'phone' in columns_map:
        phone_col = columns_map['phone']
        # 检查重复的用户编号
        duplicate_rows = df[df.duplicated(subset=[phone_col], keep=False)]
        if not duplicate_rows.empty:
            results['is_valid'] = False
            results['message'] = f'发现 {len(duplicate_rows)} 条重复用户记录'
            results['duplicate_users'] = duplicate_rows[phone_col].unique().tolist()
            results['valid_records'] -= len(duplicate_rows)
    else:
        print("警告: 未找到用户编号列，跳过唯一性检查")
    
    # 2. 检查是否存在模拟有线电视记录
    has_analog_tv = False
    invalid_indices = []
    
    if 'product' in columns_map:
        product_col = columns_map['product']
        try:
            mask = df[product_col].astype(str).str.contains('模拟有线电视', na=False)
            has_analog_tv = mask.any()
            if has_analog_tv:
                invalid_indices.extend(df[mask].index.tolist())
        except Exception as e:
            print(f"警告: 检查模拟有线电视时出错: {e}")
    else:
        print("警告: 未找到产品名称列，跳过模拟有线电视检查")
    
    if has_analog_tv:
        results['is_valid'] = False
        results['message'] = f'发现 {len(invalid_indices)} 条模拟有线电视记录'
        results['invalid_records'].extend(invalid_indices)
        results['valid_records'] -= len(invalid_indices)
    
    # 3. 检查时间有效性
    if 'time' in columns_map:
        time_col = columns_map['time']
        try:
            # 检查是否为datetime类型
            if not pd.api.types.is_datetime64_any_dtype(df[time_col]):
                # 尝试转换
                df[time_col] = pd.to_datetime(df[time_col], errors='coerce')
            
            # 检查无效时间
            invalid_time_mask = df[time_col].isna()
            if invalid_time_mask.any():
                time_invalid_indices = df[invalid_time_mask].index.tolist()
                results['is_valid'] = False
                results['message'] += f'，发现 {len(time_invalid_indices)} 条无效时间记录'
                results['invalid_records'].extend(time_invalid_indices)
                results['valid_records'] -= len(time_invalid_indices)
        except Exception as e:
            print(f"警告: 时间验证出错: {str(e)}")
    else:
        print("警告: 未找到时间列，跳过时间有效性检查")
    
    # 输出验证结果
    print(f"数据质量验证结果:")
    print(f"- 总记录数: {results['total_records']}")
    print(f"- 有效记录数: {results['valid_records']}")
    print(f"- 重复用户数: {len(results['duplicate_users'])}")
    print(f"- 无效记录数: {len(results['invalid_records'])}")
    print(f"- 验证状态: {'通过' if results['is_valid'] else '未通过'}")
    
    return results